package window;

import bean.SensorReading;
import org.apache.commons.collections.IteratorUtils;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.elasticsearch.search.aggregations.bucket.geogrid.InternalGeoHashGrid;

public class WindowTest1_TimeWindow {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<String> inputStream = env.socketTextStream("192.168.45.132", 7777);

        SingleOutputStreamOperator<SensorReading> dataStream = inputStream.map(line -> {
            String[] fields = line.split(",");
            return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
        });


        //开窗测试

        // 1. 增量聚合函数 AggregateFunction<输入, 累加器, 输出>
        //    createAccumulator() 创建一个新的累加器，启动一个新的聚合,负责迭代状态的初始化
        //    add() 对于数据的每条数据，和迭代数据的聚合的具体实现
        //    merge() 累加器
        //    getResult() 从累加器获取聚合结果
        SingleOutputStreamOperator<Integer> resultStream = dataStream.keyBy("id")
                .window(TumblingEventTimeWindows.of(Time.seconds(10)))
                .aggregate(new AggregateFunction<SensorReading, Integer, Integer>() {
                    @Override
                    public Integer createAccumulator() {
                        return 0;
                    }
                    @Override
                    public Integer add(SensorReading value, Integer accumulator) {
                        return accumulator + 1;
                    }
                    @Override
                    public Integer getResult(Integer accumulator) {
                        return accumulator;
                    }
                    @Override
                    public Integer merge(Integer a, Integer b) {
                        return a + b;
                    }
                });

        // 2. 全窗口函数
        SingleOutputStreamOperator<Tuple3<String, Long, Integer>> resultStream2 = dataStream.keyBy("id")
                .window(TumblingEventTimeWindows.of(Time.seconds(10)))
                .apply(new WindowFunction<SensorReading, Tuple3<String, Long, Integer>, Tuple, TimeWindow>() {
                    @Override
                    public void apply(Tuple tuple, TimeWindow timeWindow, Iterable<SensorReading> iterable, Collector<Tuple3<String, Long, Integer>> collector) throws Exception {
                        String id = tuple.getField(0);
                        long windowEnd = timeWindow.getEnd();
                        int count = IteratorUtils.toList(iterable.iterator()).size();
                        collector.collect(new Tuple3<>(id, windowEnd, count));
                    }
                });


        resultStream.print();
        env.execute();

    }
}
